6장. 랭그래프로 설계하는 RAG 파이프라인

출처 — 브라이스 유·조경아·박수진·김재웅, 『RAG 마스터: 랭체인으로 완성하는 LLM 서비스』(프리렉, 2025), 6장 (pp. 367~426). 원문 PDF rag_master_final_v11_260910.pdf (2026-09-10 판)

랭그래프는 노드·에지·상태로 이루어진 그래프 구조로 LLM 파이프라인의 순환과 분기를 다루며, 이 장은 그 구성요소부터 실전 챗봇·자체교정 RAG·코드 어시스트 챗봇까지 단계적으로 구현한다.

학습 목표

이 장을 끝내면 다음을 할 수 있다. - 그래프·상태·노드·에지 네 구성요소의 역할과 관계를 설명한다 - 조건부 에지로 도구 호출 여부를 분기하는 챗봇을 구현한다 - 체크포인터로 대화 상태를 저장하고, 정적 중단점과 동적 중단으로 루프에 개입하는 방법을 구분한다 - 자체교정 RAG 와 코드 어시스트 챗봇의 그래프 구조를 각각 설계한다 - 이 장의 "그래프"(랭그래프의 자료구조)와 5장의 "그래프 RAG"(지식 그래프)가 서로 다른 개념임을 구분한다

전체 흐름도

[1. 랭그래프의 구성요소]
그래프(노드+에지) → 상태(State, 리듀서) → 노드(START/END) → 에지(일반/조건부, 진입 지점)
                                                                    │
                                                                    ▼
[2. 랭그래프로 챗봇 만들기] — 같은 챗봇에 절마다 기능을 하나씩 추가한다
2.1 루프 구현하기 → 2.2 조건문 구현하기(도구 호출 분기) → 2.3 스트리밍(values/updates)
                                                                    │
                                                                    ▼
2.4 상태 저장하기(체크포인터·thread_id) → 2.5 루프 개입하기(interrupt_before)
                                                                    │
                                                                    ▼
[3. 랭그래프 실습] — 구성요소를 실전 RAG·코드 생성에 응용한다
3.1 자체교정 RAG
   검색 → 문서 평가
          ├─ 관련 있음 → 답변 생성
          └─ 관련 없음 → 질문 재작성 → 웹 검색 → 답변 생성
3.2 코드 어시스트 챗봇
   코드 생성 → 코드 검사(import·실행)
          ├─ 오류 있음(3회 미만) → 오류 반영(reflect) → 코드 생성(재시도)
          └─ 오류 없음 또는 3회 도달 → 종료

0. 용어 사전

참고 — 위쪽 4개는 이 장을 읽기 전에 알아야 하는 선행 용어다. 낯설면 1장(랭체인 기초)·2장(RAG 기초)을 먼저 본다.

1. 랭그래프의 구성요소

랭그래프는 LLM 기반 에이전트 시스템의 순환과 분기를 그래프 구조로 다루는 라이브러리다. "생성된 답변이 충분한지 재생성할지"·"어떤 도구를 호출할지"처럼 LLM 이 스스로 판단해야 하는 분기점이 많아질수록 파이프라인은 복잡해지는데, 랭그래프는 이런 분기점을 노드와 에지로 표현해 구현을 단순화한다.

1.1 그래프

그래프는 객체 간의 관계를 나타내는 자료구조로, 노드(개별 객체)와 에지(노드 간 관계·경로)로 이루어진다. 랭그래프의 그래프는 구글의 대규모 그래프 처리 시스템 프리겔에서 영감을 받은 슈퍼스텝 방식으로 동작한다. 슈퍼스텝은 여러 노드가 동시에 자신의 작업을 수행하는 한 단위다 — 한 노드가 끝난 뒤 다음 노드가 시작하는 것이 아니라 여러 노드가 병렬로 움직이며, 동시에 실행되는 노드는 같은 슈퍼스텝에, 순차적으로 실행되는 노드는 별도의 슈퍼스텝에 속한다. 노드는 하나 이상의 입력 에지에서 새 메시지(상태)를 받으면 활성화되어 자신의 기능을 실행하고, 각 슈퍼스텝이 끝날 때 입력 메시지가 없는 노드는 비활성화로 표시된다. 모든 노드가 비활성화되고 더 이상 메시지가 전송되지 않으면 그래프 실행이 종료된다.

랭그래프에는 두 가지 그래프 클래스가 있다. 상태 그래프(StateGraph) 는 사용자가 정의하는 상태를 매개변수로 쓰는 일반적인 클래스이고, 메시지 그래프(MessageGraph) 는 메시지 목록만으로 이루어지는 특별한 클래스로 주로 챗봇 같은 대화형 시스템에 쓰인다.

1.2 상태

그래프를 정의할 때 가장 먼저 할 일은 상태를 정의하는 것이다. 상태는 애플리케이션 내에서 메시지로 주고받는 변수들의 집합이며, 파이썬의 모든 타입으로 정의할 수 있지만 대체로 TypedDict 나 Pydantic 의 BaseModel 로 선언한다. 상태는 그래프 내 모든 노드와 에지의 입력으로 쓰이고, 각 노드는 상태를 업데이트할 수 있다.

from typing import TypedDict

class State(TypedDict):
    count: int
    messages: list[str]

이 예시는 count 와 messages 두 필드를 갖는 상태 클래스다. 리듀서 를 쓰면 기존 상태에 새 업데이트를 결합해 새로운 상태를 만들 수 있다. Annotated 타입으로 리듀서 함수를 지정하면 그 필드는 리듀서를 거쳐 업데이트된다.

from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    count: int
    messages: Annotated[list[str], add]

이 예시에서는 messages 필드에 add 리듀서가 지정돼, 새 메시지가 들어올 때마다 기존 리스트와 병합된다. 리듀서를 지정하지 않으면 새 값이 기존 값을 그대로 덮어쓴다는 점이 중요하다 — 대화 이력처럼 누적돼야 하는 필드에 리듀서를 빠뜨리면 매 턴마다 이전 메시지가 사라진다.

1.3 노드

노드는 실제 작업을 수행하는 실행 단위다. 에이전트의 로직을 담은 파이썬 함수가 곧 노드이며, 상태를 입력으로 받아 동작하고 그 결과로 상태값을 업데이트해 반환한다. 노드는 첫 번째 인자로 상태값(state)을, 두 번째 인자로 설정값(config)을 받을 수 있고, add_node() 로 그래프에 추가한다.

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

# 상태 그래프 선언
builder = StateGraph(dict)

# 노드로 사용할 함수 정의
def my_node(state: dict, config: RunnableConfig):
    print("In node: ", config["configurable"]["user_id"])
    return {"results": f"Hello, {state['input']}!"}

def my_other_node(state: dict):
    return state  # 상태를 그대로 반환

# 노드를 그래프에 추가
builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)

my_node 는 입력된 상태값을 활용해 인사말을 반환하고, my_other_node 는 상태값을 변경하지 않고 그대로 반환하는 노드다.

START 노드 는 그래프 실행의 시작점을 나타내는 특별한 노드로, 사용자 입력을 처음 받아 그래프로 전달하며 그래프 내에서 첫 번째로 실행될 노드를 지정할 때 쓴다.

from langgraph.graph import START
graph.add_edge(START, "node_a")

END 노드 는 그래프 실행이 완료됐음을 나타내는 종료 노드다. 특정 노드의 작업이 끝난 후 더 처리할 작업이 없으면 END 노드로 연결해 실행을 종료한다.

from langgraph.graph import END
graph.add_edge("node_a", END)

1.4 에지

에지는 노드가 작업을 마친 뒤 다음에 어떤 동작을 이어갈지 결정하는 흐름 제어 요소다. 파이썬 함수나 고정된 연결로 다음 실행 노드를 지정하며, 조건에 따라 분기하거나 종료를 지시할 수도 있다. 하나의 노드는 여러 에지를 가질 수 있다.

일반 에지 는 가장 기본적인 형태로, 한 노드에서 다음 노드로 직접 이동할 때 쓴다.

graph.add_edge("node_a", "node_b")

이 코드는 node_a 의 작업이 끝난 후 node_b 가 실행되도록 지정한다.

조건부 에지 는 특정 조건에 따라 다른 노드로 분기하거나 종료할 때 쓴다. 조건을 판단하는 함수가 필요하며, 그 반환값에 따라 다음 실행 노드를 선택한다.

graph.add_conditional_edges(
    "node_a",
    routing_function,
    {True: "node_b", False: "node_c"},
)

routing_function 이 node_a 다음에 어떤 노드를 쓸지 결정하는 함수가 되며, 반환값이 True 이면 node_b, False 이면 node_c 가 다음 실행 노드가 된다.

진입 지점(Entry Point) 은 그래프가 시작될 때 처음 실행할 노드를 명시한다. 주로 START 라는 가상의 노드를 써서 첫 실행 지점을 설정한다.

from langgraph.graph import START
graph.add_edge(START, "node_a")

조건부 진입 지점 은 사용자 입력이나 외부 조건에 따라 첫 번째 노드를 동적으로 결정한다.

from langgraph.graph import START
graph.add_conditional_edges(
    START,
    routing_function,
    {True: "node_b", False: "node_c"},
)

add_conditional_edges() 는 가상의 START 노드와 라우팅 함수를 입력받고, 세 번째 인자로 라우팅 함수의 반환값에 해당하는 노드 매핑 정보를 받아 조건부로 진입 지점 노드를 선택한다.

2. 랭그래프로 챗봇 만들기

이제 랭그래프로 오픈AI LLM 기반 챗봇을 단계별로 구현한다. 이 챗봇은 (1) 웹 검색으로 최신 정보에 답하고, (2) 이전 대화와 사용자 설정을 저장해 맥락을 유지하고, (3) 복잡한 질문은 사람에게 라우팅하고, (4) 커스텀 상태값으로 동작을 유연하게 제어하고, (5) 이전 대화로 되돌아가 수정하는 기능을 포함한다. 실습 코드는 책의 깃허브 저장소 6장 폴더의 ch06_LANG_GRAPH.ipynb 파일이다.

먼저 구글 코랩에서 패키지를 설치하고 환경 변수를 로드한다.

%%capture --no-stderr
%pip install -U langgraph
%pip install -U langchain-openai

from google.colab import drive
drive.mount('/content/drive')

from dotenv import load_dotenv
# 오픈AI·Tavily API 키가 담긴 .env 를 읽어 환경 변수로 등록
load_dotenv("/content/.env")

이어서 그래프의 상태값을 정의한다. 상태는 대화 메시지를 포함하며, 랭그래프의 add_messages 리듀서로 메시지를 누적한다.

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 실습에서 사용할 그래프의 상태 정의
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 실습에서 사용할 그래프 인스턴스 생성
graph_builder = StateGraph(State)

2.1 루프 구현하기

챗봇 그래프의 기본 루프부터 구현한다. 먼저 LLM 이 사용자 질의를 받아 응답을 생성하는 챗봇 노드를 추가한다.

from langchain_openai import ChatOpenAI

# 오픈AI 클라이언트 정의
llm = ChatOpenAI(model="gpt-4o-mini")

# 오픈AI를 호출하여 응답을 받아온 뒤, 상태값에 저장하여 반환하는 챗봇 함수 정의
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 챗봇 노드 정의
graph_builder.add_node("chatbot", chatbot)

다음으로 진입 지점과 종료 지점을 지정하고, 그래프를 컴파일해 실행 가능한 형태로 만든다. 컴파일 이란 정의한 노드와 흐름을 실제로 실행할 수 있는 구조로 바꾸는 것을 뜻한다.

from langgraph.graph import StateGraph, START, END

# 진입 지점
graph_builder.add_edge(START, "chatbot")
# 종료 지점
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()

사용자 입력을 받아 순환하며 동작하는 챗봇은 while 루프로 구현한다. 사용자가 quit·exit·q 를 입력할 때까지 질문과 응답을 반복하며, 입력은 graph.stream() 을 통해 그래프에 전달된다.

while True:
    user_input = input("User: ")
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break
    for event in graph.stream({"messages": [("user", user_input)]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

실행 예시:

User: 너는 누구야?
Assistant: 저는 인공지능 챗봇이에요. 무엇을 도와드릴까요?
User: 반가워!
Assistant: 안녕하세요! 만나서 반가워요! 어떻게 도와드릴까요?
User: exit
Goodbye!

마지막으로 draw_mermaid_png() 로 그래프 구조를 이미지로 시각화할 수 있다.

from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

지금까지 구성한 그래프는 START → chatbot → END 로 이어지는 단순한 직선 흐름이다.

2.2 조건문 구현하기

이제 챗봇이 학습된 지식만으로 답할 수 없는 질문에도 대응하도록 외부 검색 도구를 연동한다. 이번 예제는 Tavily 검색 엔진을 챗봇의 도구로 써서 실시간 정보를 제공한다. Tavily 는 웹에서 데이터를 수집하고 검색어와 연관성이 높은 정보를 정형화된 JSON 형식으로 제공하는 검색 API 다. 사용하려면 tavily.com 에서 회원 가입(또는 구글·깃허브 계정으로 로그인)하고, 대시보드(app.tavily.com/home)에서 API 키를 발급받아 .env 파일에 TAVILY_API_KEY 로 저장한다(2026-09-12 기준 이 대시보드 주소는 정상 동작한다 — 현재 계약 방식은 아래 최신 동향 참고).

%%capture --no-stderr
%pip install -U tavily-python
%pip install -U langchain_community

Tavily 는 랭체인 라이브러리로 제공되므로 패키지 설치 후 코드 한 줄로 도구를 정의할 수 있다.

from langchain_community.tools.tavily_search import TavilySearchResults

# Tavily 검색 엔진을 도구로 정의
tool = TavilySearchResults(max_results=2)
tools = [tool]

# 호출 예시
tool.invoke("내일 대한민국 서울의 날씨는?")

정의한 도구는 llm.bind_tools(tools) 로 LLM 과 연결한다. 이제 LLM 은 질문 유형에 따라 검색 도구 사용 여부를 판단하고, 필요한 파라미터를 포함한 응답을 반환한다.

from typing import Annotated
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

챗봇은 이제 두 상황에 대응해야 한다 — LLM 이 도구 호출이 필요하다고 판단하면 도구를 호출해야 하고, 단순 답변이면 사용자에게 바로 반환하고 종료해야 한다. 도구 호출 정보(tool_calls)가 있는 최근 메시지를 검사해 도구를 실행하는 노드를 직접 만들어 본다.

import json
from langchain_core.messages import ToolMessage

class BasicToolNode:
    """최근 AI 메시지의 tool_calls 를 실행하는 노드"""

    def __init__(self, tools: list) -> None:
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No message found in input")
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result, ensure_ascii=False),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}

# 도구 노드 정의
tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

도구 노드는 LLM 응답에 tool_calls 가 포함된 경우에만 활성화돼야 한다. hasattr(ai_message, "tool_calls") 와 len(ai_message.tool_calls) > 0 조건으로 이를 검사해, 있으면 "tools" 를, 없으면 종료를 뜻하는 값을 반환하는 라우팅 함수를 만든다.

from typing import Literal

def route_tools(state: State) -> Literal["tools", "__end__"]:
    if isinstance(state, list):
        ai_message = state[-1]
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return "__end__"

# 챗봇 노드에 조건부 에지를 정의
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", "__end__": "__end__"},
)

도구 노드가 실행된 후에는 다시 챗봇 노드로 이동해 검색 결과를 바탕으로 최종 응답을 생성한다. 도구 노드와 챗봇 노드를 연결하고, 진입 지점으로 챗봇 노드를 지정한다.

graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

구성도를 보면 챗봇의 응답에 따라 도구를 호출하거나, 직접 답변하고 종료하는 두 흐름을 확인할 수 있다.

참고 — 프리빌트 컴포넌트(Prebuilt Components) 랭그래프는 자주 쓰는 기능·패턴을 미리 구현해 둔 프리빌트 컴포넌트를 제공한다. 위에서 직접 만든 BasicToolNode 와 route_tools 는 매우 전형적인 패턴이라 랭그래프가 각각 ToolNode·tools_condition 으로 미리 만들어 두었다 — import 만으로 손쉽게 대체할 수 있다(자세한 대체 코드는 §2.4). 참고: prebuilt 레퍼런스

2.3 스트리밍

랭그래프는 실시간으로 데이터를 지속적으로 전송하는 스트리밍 을 지원한다. 대규모 작업이 진행 중일 때도 중간 결과를 즉시 확인할 수 있고, 전체 결과가 준비되기 전에도 사용자와 상호작용할 수 있다. 스트리밍은 graph.stream() 호출 시 stream_mode 파라미터로 동작 방식을 고른다. values 는 각 노드 실행 후 그래프의 전체 상태를 반환해 워크플로우 전체 상태 변화를 추적하기 좋고, updates 는 각 노드 실행 후 변경된 부분만 반환해 변경 지점만 빠르게 확인하기 좋다.

from langchain_core.messages import BaseMessage

while True:
    user_input = input("User: ")
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break
    events = graph.stream(
        input={"messages": [("user", user_input)]},
        stream_mode="updates",
    )
    for event in events:
        for value in event.values():
            if isinstance(value["messages"][-1], BaseMessage):
                print("Assistant:", value["messages"][-1].content)

updates 모드로 "24년 9월 9일 서울 날씨"를 물으면, Tavily 검색 결과(뉴스 URL과 요약)가 먼저 한 번 출력되고, 이어서 그 결과를 종합한 최종 답변이 출력된다 — 매번 상태가 바뀔 때마다 그 시점의 마지막 메시지만 보여 주므로, 전체 대화 내역이 아니라 최신 응답만 확인하게 된다는 점이 values 모드와의 차이다.

2.4 상태 저장하기

지금까지의 챗봇은 최신 정보를 검색할 수 있지만, 이전 질문과 답변의 맥락을 기억하지 못해 멀티턴 대화를 자연스럽게 이어가지 못한다. 랭그래프는 그래프 상태를 메모리나 데이터베이스 같은 외부 저장소에 저장하고 이후 복원하는 체크포인트(checkpoint) 기능을 제공한다. 그래프를 컴파일할 때 데이터를 저장할 체크포인터(checkpointer) 를 설정하고, 호출할 때 thread_id 를 함께 전달하면 이전 대화 상태를 불러와 이어갈 수 있다. 가장 간단한 형태인 MemorySaver 를 쓴다.

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()

프리빌트 컴포넌트 ToolNode·tools_condition 으로 §2.2 의 BasicToolNode·route_tools 를 대체하고, checkpointer 를 지정해 컴파일한다.

from typing import Annotated
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

class State(TypedDict):
    messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

# 미리 빌드된 도구 노드
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)

# 미리 빌드된 조건부 에지
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

# 체크포인터를 지정하여 그래프를 컴파일
graph = graph_builder.compile(checkpointer=memory)

ToolNode 는 직접 만든 BasicToolNode 와 동일하게 동작하고, tools_condition 은 route_tools 와 같은 역할을 한다 — 코드가 훨씬 간결해지고 구현이 직관적이다.

대화의 키로 쓸 thread_id 를 config 형식으로 정의하고, graph.stream() 호출 시 함께 전달한다.

config = {"configurable": {"thread_id": "1"}}

user_input = "안녕! 내 이름은 오해원이야."
events = graph.stream(
    {"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
    event["messages"][-1].pretty_print()

Human Message
안녕! 내 이름은 오해원이야.
Ai Message
안녕하세요, 오해원님! 만나서 반갑습니다. 어떻게 도와드릴까요?

같은 thread_id 로 "내 이름을 기억하니?" 를 물으면 챗봇은 이름을 기억한다. 반면 thread_id 를 "2" 로 바꿔 같은 질문을 하면, 그 스레드에는 저장된 맥락이 없으므로 챗봇은 이름을 기억하지 못한다 — 체크포인트는 thread_id 단위로 분리된다.

graph.get_state(config) 로 현재 그래프 상태의 스냅샷(StateSnapshot)을 조회하면, 해당 thread_id 에 저장된 모든 메시지·상태값·파라미터 히스토리를 확인할 수 있다.

2.5 루프 개입하기

에이전트의 행동을 신뢰할 수 없어 작업을 검토하거나 승인해야 하거나, 그래프 실행을 수동으로 중단하고 흐름을 수정해야 할 때가 있다. 이런 인간 개입(human-in-the-loop) 은 그래프를 컴파일하는 compile() 메서드의 interrupt_before 파라미터에 개입하고자 하는 노드를 명시해 구현한다 — 그 노드를 실행하기 직전에 흐름이 멈춘다.

graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"],
)

"지금 서울 날씨 어때?" 를 입력하면, 도구 호출이 필요한 상태에서 실행이 멈춘다 — AI 메시지에 응답 내용 대신 도구 호출 파라미터만 담겨 있다.

user_input = "지금 서울 날씨 어때?"
config = {"configurable": {"thread_id": "2"}}
events = graph.stream(
    {"messages": [("user", user_input)]}, config, stream_mode="values"
)
for event in events:
    if "messages" in event:
        event["messages"][-1].pretty_print()

graph.get_state(config).next 로 다음 실행될 노드를 확인하면 ("tools",) 가 나온다 — 그래프가 tools 노드 앞에서 멈춰 있다는 뜻이다. 이 시점에서 상태를 자유롭게 편집할 수 있다. 예를 들어 실제 도구를 호출하지 않고 임의의 응답을 강제로 넣어 본다.

from langchain_core.messages import AIMessage

existing_message = snapshot.values["messages"][-1]
existing_message_id = existing_message.tool_calls[0]["id"]

answer = "서울의 날씨는 매우 맑아요."
new_messages = [
    ToolMessage(content=answer, tool_call_id=existing_message_id),
    AIMessage(content=answer),
]
graph.update_state(config, {"messages": new_messages})

update_state() 로 새 메시지를 추가하는 대신, 기존 메시지를 직접 수정할 수도 있다. 예를 들어 도구 호출의 query 인자를 가로채 다른 질의로 바꿔치기한다.

snapshot = graph.get_state(config)
existing_message = snapshot.values["messages"][-1]
new_tool_call = existing_message.tool_calls[0].copy()
new_tool_call["args"]["query"] = "지금 경기도 날씨 어때?"
new_message = AIMessage(
    content=existing_message.content,
    tool_calls=[new_tool_call],
    id=existing_message.id,
)
graph.update_state(config, {"messages": [new_message]})

실제 사용자가 입력한 메시지는 "지금 서울 날씨 어때?" 지만, 개입 후 도구가 실제로 검색하는 질의는 "지금 경기도 날씨 어때?" 로 바뀐다 — 사람이 그래프 실행 중 언제든 개입해 흐름과 데이터를 원하는 방식으로 바꿀 수 있음을 보여 준다. 루프 개입으로는 현재 상태 편집, 과거 기록 탐색, 상태 수정, 특정 시점에 메시지 추가 같은 작업이 가능하다(정적 중단점의 현재 위상은 최신 동향 참고).

3. 랭그래프 실습

이 절에서는 랭그래프의 구성요소를 실제 애플리케이션에 응용한다.

3.1 자체교정 RAG

기본적인 RAG 는 사용자 질문에 대해 관련 문서를 검색하고 이를 기반으로 답변을 생성하는데, 검색된 문서가 질문과 충분히 관련 없으면 답변 품질이 크게 떨어진다. 자체교정 RAG(Corrective-RAG) 는 질문과 검색된 문서의 연관도를 평가한 뒤, 연관도가 기준 이하면 질문을 재작성해 웹 검색으로 검색을 다시 수행함으로써 더 적합한 문서를 확보한다. 구현 순서는 (1) 질문 입력과 문서 검색 (2) 문서와 질문의 연관성 평가 (3) 연관성이 낮으면 웹 검색으로 정보 보완 (4) 웹 검색 전 질문을 검색에 적합한 형태로 변형, 이렇게 네 단계다. 실습 코드는 ch06_LANG_GRAPH_CORRECTIVE_RAG.ipynb 파일이다.

환경 설정

!pip install langchain_community tiktoken langchain-openai chromadb langchain langgraph tavily-python

from google.colab import drive
drive.mount('/content/drive')

from dotenv import load_dotenv
load_dotenv("/content/.env")

문서 인덱싱 — 사용자 질문을 처리하기 전에 검색 대상 문서를 인덱싱해야 한다. 이번 실습은 구글의 코드 스타일 가이드 문서를 크롤링해 벡터 데이터베이스를 구축한다.

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

urls = [
    "https://google.github.io/styleguide/pyguide.html",
    "https://google.github.io/styleguide/javaguide.html",
    "https://google.github.io/styleguide/jsguide.html",
]
docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=250, chunk_overlap=0
)
doc_splits = text_splitter.split_documents(docs_list)

vectorstore = Chroma.from_documents(
    documents=doc_splits,
    collection_name="rag-chroma",
    embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()

WebBaseLoader 로 지정한 URL 의 웹 페이지를 크롤링하고, 긴 문서를 그대로 검색하면 느리므로 RecursiveCharacterTextSplitter 로 250 토큰 단위로 분할한다. 분할된 조각을 Chroma 벡터 저장소에 OpenAIEmbeddings 로 임베딩해 저장하고, 검색을 수행할 retriever 를 만든다.

문서 평가하기 — 검색된 문서가 질문과 얼마나 관련 있는지 LLM 으로 평가하는 노드를 만든다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

class GradeDocuments(BaseModel):
    binary_score: str = Field(
        description="문서와 질문의 연관성 여부. (예 or 아니오)"
    )

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm_grader = llm.with_structured_output(GradeDocuments)

system = """당신은 사용자의 질문에 대해 검색된 문서의 관련성을 평가하는 전문가입니다.
문서에 질문과 관련된 키워드나 의미가 담겨 있으면, 해당 문서를 '관련 있음'으로 평가하세요.
문서가 질문과 관련이 있는지 여부를 '예' 또는 '아니오'로 표시해 주세요."""
grade_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "검색된 문서: \n\n {document} \n\n 사용자 질문: {question}"),
    ]
)
retrieval_grader = grade_prompt | structured_llm_grader

GradeDocuments 데이터 모델로 평가 결과(예/아니오)를 구조화하고, with_structured_output() 으로 LLM 응답이 이 형식을 따르게 한다.

question = "파이썬 코드 작성 가이드"
docs = retriever.invoke(question)
doc_txt = docs[1].page_content
print(retrieval_grader.invoke({"question": question, "document": doc_txt}))

binary_score='예'

답변 생성하기 — 검색된 문서가 적절하면 그 문서를 맥락으로 답변을 생성한다.

from langchain_core.output_parsers import StrOutputParser

system = """당신은 질문에 답변하는 업무를 돕는 도우미입니다.
제공된 문맥을 바탕으로 질문에 답변하세요. 만약 답을 모르면 모른다고 말하세요.
세 문장을 넘지 않도록 답변을 간결하게 작성하세요."""
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "질문: {question} \n문맥: {context} \n답변:"),
    ]
)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = prompt | llm | StrOutputParser()

prompt | llm | StrOutputParser() 로 프롬프트·LLM·문자열 출력을 결합한 LCEL 체인 rag_chain 을 만든다. 여기에 질문과 문맥을 넘기면 답변을 생성한다.

질문 재작성하기 — 검색된 문서가 적절하지 않으면 웹 검색을 해야 하는데, 이때 질문을 웹 검색에 적합한 형태로 변환한다.

system = """당신은 입력된 질문을 변형하여 웹 검색에 최적화된 형태로 만드는 질문 생성기입니다.
입력된 질문을 보고 그 이면에 있는 의미나 의도를 파악해주세요."""
re_write_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("human", "질문: \n\n {question} \n 더 나은 질문으로 바꿔주세요."),
    ]
)
question_rewriter = re_write_prompt | llm | StrOutputParser()

question = "C++ 깔끔하게 짜고 싶다"
question_rewriter.invoke({"question": question})

'어떻게 하면 C++ 코드를 더 깔끔하게 작성할 수 있을까요?'

웹 검색하기 — TavilySearchResults 로 웹 검색 도구를 정의한다.

from langchain_community.tools.tavily_search import TavilySearchResults
web_search_tool = TavilySearchResults(k=3)

상태 — 그래프를 작성하기 전에 상태값을 정의한다.

from typing import List
from typing_extensions import TypedDict

class GraphState(TypedDict):
    question: str        # 사용자 질문 또는 재작성된 질문
    generation: str       # LLM이 생성한 답변
    web_search: str       # 웹 검색 필요 여부
    documents: List[str]  # 검색된 문서 리스트

그래프 — 노드와 에지로 Corrective-RAG 를 완성한다.

from langchain.schema import Document

def retrieve(state):
    """문서를 검색합니다."""
    print("---검색---")
    question = state["question"]
    documents = retriever.get_relevant_documents(question)
    return {"documents": documents, "question": question}

def generate(state):
    """답변을 생성합니다."""
    print("---생성---")
    question = state["question"]
    documents = state["documents"]
    generation = rag_chain.invoke({"context": documents, "question": question})
    return {"documents": documents, "question": question, "generation": generation}

def grade_documents(state):
    """검색된 문서가 질문과 연관이 있는지 평가합니다."""
    print("---문서와 질문의 연관성 평가---")
    question = state["question"]
    documents = state["documents"]

    filtered_docs = []
    web_search = "아니오"
    for d in documents:
        score = retrieval_grader.invoke(
            {"question": question, "document": d.page_content}
        )
        grade = score.binary_score
        if grade == "예":
            print("---평가: 연관 문서---")
            filtered_docs.append(d)
        else:
            print("---평가: 연관 없는 문서---")
            web_search = "예"
            continue
    return {"documents": filtered_docs, "question": question, "web_search": web_search}

def transform_query(state):
    """질문을 더 적합한 형태로 변환합니다."""
    print("---질문 변환---")
    question = state["question"]
    documents = state["documents"]
    better_question = question_rewriter.invoke({"question": question})
    return {"documents": documents, "question": better_question}

def web_search(state):
    """웹 검색을 수행합니다."""
    print("---웹 검색---")
    question = state["question"]
    documents = state["documents"]
    docs = web_search_tool.invoke({"query": question})
    web_results = "\n".join([d["content"] for d in docs])
    web_results = Document(page_content=web_results)
    documents.append(web_results)
    return {"documents": documents, "question": question}

노드마다 상태값을 입력받아 미리 정의한 체인을 실행하고, 새 상태값을 만들어 반환한다. 다음은 노드를 연결할 에지 함수다.

def decide_to_generate(state):
    """답변을 생성할지, 질문을 재생성할지 결정합니다."""
    print("---문서 검토---")
    web_search = state["web_search"]
    if web_search == "예":
        print("---연관 문서가 없음. 질문을 변환---")
        return "transform_query"
    else:
        print("---연관 문서가 있음. 답변을 생성---")
        return "generate"

decide_to_generate 는 grade_documents 가 세운 web_search 플래그를 검사해 다음에 호출할 노드를 결정한다. 이제 노드와 에지를 그래프에 추가해 워크플로우를 구성한다.

from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search_node", web_search)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "transform_query": "transform_query",
        "generate": "generate",
    },
)
workflow.add_edge("transform_query", "web_search_node")
workflow.add_edge("web_search_node", "generate")
workflow.add_edge("generate", END)

app = workflow.compile()

완성된 그래프를 실행해 확인한다. 먼저 관련 문서가 충분한 질문을 입력한다.

inputs = {"question": "구글의 코드 작성 가이드"}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint(f"Node '{key}':")
        pprint(value["generation"])

---검색---
---문서와 질문의 연관성 평가---
---평가: 연관 문서---(4회)
---문서 검토---
---연관 문서가 있음. 답변을 생성---
---생성---
('구글의 코드 작성 가이드는 JavaScript와 Java 프로그래밍 언어에 대한 코딩 표준을 정의합니다. ...')

관련 문서가 충분하면 검색된 문서만으로 바로 답변을 생성한다. 이번에는 인덱싱한 문서와 무관한 질문을 넣는다.

inputs = {"question": "C++ 깔끔하게 짜고 싶다"}
for output in app.stream(inputs):
    for key, value in output.items():
        pprint(f"Node '{key}':")
        pprint(value["generation"])

---검색---
---문서와 질문의 연관성 평가---
---평가: 연관 없는 문서---(2회)
---평가: 연관 문서---(2회)
---문서 검토---
---연관 문서가 없음. 질문을 변환---
---질문 변환---
---검색---(웹 검색)
---생성---
('C++ 코드를 깔끔하고 효율적으로 작성하려면, 가독성을 높이기 위해 주석을 적절히 추가하고, ...')

연관 없는 문서가 있으면 질문을 재작성하고 웹 검색 결과를 추가해 답변을 생성한다. 이렇게 답변 품질을 스스로 평가하고 개선하는 Corrective-RAG 가 완성됐다.

3.2 코드 어시스트 챗봇

코파일럿처럼 LLM 으로 코드를 생성해 개발을 돕는 챗봇을 만든다. 동작 순서는 (1) 사용자가 질문과 코드 맥락을 제공 (2) 코드 맥락을 분석해 답변을 생성 (3) 구조화된 출력을 위한 도구를 호출 (4) 최종 답변을 반환하기 전에 import·코드 실행 두 가지 단위 테스트를 수행, 이렇게 네 단계다. 실습 코드는 ch06_LANG_GRAPH_CODE_ASSIST_CHATBOT.ipynb 파일이다.

환경 설정

%%capture --no-stderr
!pip install langchain_community tiktoken langchain-openai chromadb langchain langgraph

from google.colab import drive
drive.mount('/content/drive')

from dotenv import load_dotenv
load_dotenv("/content/.env")

문서 정의 — LLM 에 제공할 코드 맥락으로 랭체인 LCEL 공식 문서를 크롤링한다. 원문이 크롤링 대상으로 쓴 옛 주소(python.langchain.com/v0.2/..., 각주 2)는 현재 일반 개요 페이지로 리다이렉트되므로, 아래 코드는 그 자리에 현재 랭체인 공식 문서의 루트 경로를 넣었다(생존 확인 2026-09-12). 크롤링 대상은 실행 시점의 최신 공식 문서 경로로 다시 확인하는 것이 안전하다.

from bs4 import BeautifulSoup as Soup
from langchain_community.document_loaders.recursive_url_loader import RecursiveUrlLoader

# 원문의 v0.2 LCEL 개념 문서 경로가 이사해, 현재 랭체인 공식 문서 루트로 대체했다
url = "https://docs.langchain.com/oss/python/langchain/overview"
loader = RecursiveUrlLoader(
    url=url, max_depth=20, extractor=lambda x: Soup(x, "html.parser").text
)
docs = loader.load()
d_sorted = sorted(docs, key=lambda x: x.metadata["source"])
d_reversed = list(reversed(d_sorted))
concatenated_content = "\n\n\n --- \n\n\n".join(
    [doc.page_content for doc in d_reversed]
)

RecursiveUrlLoader 는 지정한 URL 을 기준으로 하위 페이지까지 재귀적으로 크롤링하며, BeautifulSoup 으로 페이지에서 텍스트만 추출한다. 이렇게 모은 문서는 LLM 이 코드 생성 질문에 답할 때 참고하는 맥락이 된다.

코드 생성 — 사용자 요청에 따라 코드를 생성하는 함수를 만든다. LLM 이 LCEL 전문가로서 답변하도록 지시하는 프롬프트를 정의하고, 출력을 구조적으로 저장할 데이터 모델 code 를 정의한다.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

system = """
당신은 LCEL(LangChain expression language) 전문가인 코딩 어시스턴트입니다.
다음은 필요한 LCEL 문서 전문입니다:
{context}
위에 제공된 문서를 기반으로 사용자 질문에 답변하세요.
제공하는 코드는 실행 가능해야 하며, 필요한 모든 import 문과 변수들이 정의되어 있어야 합니다.
답변을 다음과 같은 구조로 작성하세요:
1. prefix : 문제와 접근 방식에 대한 설명
2. imports : 코드 블록 import 문
3. code : import 문을 제외한 코드 블록
4. description : 질문에 대한 코드 스키마
다음은 사용자 질문입니다:
"""
code_gen_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system),
        ("placeholder", "{messages}"),
    ]
)

class code(BaseModel):
    prefix: str = Field(description="문제와 접근 방식에 대한 설명")
    imports: str = Field(description="코드 블록 import 문")
    code: str = Field(description="import 문을 제외한 코드 블록")
    description: str = Field(description="질문에 대한 코드 스키마")

llm = ChatOpenAI(temperature=0, model="gpt-4o-mini")
code_gen_chain = code_gen_prompt | llm.with_structured_output(code)

llm.with_structured_output(code) 로 응답을 code 데이터 모델 형식에 맞춰 구조화한다.

question = "LCEL로 RAG 체인을 어떻게 만들어?"
solution = code_gen_chain.invoke(
    {"context": concatenated_content, "messages": [("user", question)]}
)
print(solution)

응답은 prefix(설명)·imports(import 문)·code(본 코드)·description(요약)으로 구조화되어 반환된다.

상태 — 코드 어시스턴트 챗봇에서 주고받을 상태를 정의한다.

from typing import List, TypedDict

class GraphState(TypedDict):
    error: str        # 테스트 오류 발생 여부
    messages: List     # 질문·오류 메시지·이유를 포함하는 메시지 목록
    generation: str    # 생성된 코드
    iterations: int    # 시도 횟수

그래프 — 코드를 생성하고 자동으로 테스트해 검증하는 흐름을 구현한다. 코드 검증에는 파이썬의 exec() 함수를 써서, LLM 이 생성한 코드를 실제로 실행해 오류 여부를 확인한다.

def generate(state: GraphState):
    """코드를 생성합니다."""
    print("---코드 생성---")
    messages = state["messages"]
    iterations = state["iterations"]
    error = state.get("error", "no")

    if error == "yes":
        messages += [
            (
                "user",
                "다시 시도해보세요. 출력 결과를 prefix, imports, code block으로 구조화하기 위해 코드 도구를 호출하세요:",
            )
        ]

    code_solution = code_gen_chain.invoke(
        {"context": concatenated_content, "messages": messages}
    )
    messages += [
        (
            "assistant",
            f"{code_solution.prefix} \n Imports: {code_solution.imports} \n Code: {code_solution.code}",
        )
    ]
    iterations = iterations + 1
    return {"generation": code_solution, "messages": messages, "iterations": iterations}


def code_check(state: GraphState):
    """코드 검사"""
    print("---코드 검사---")
    messages = state["messages"]
    code_solution = state["generation"]
    iterations = state["iterations"]

    imports = code_solution.imports
    code = code_solution.code

    try:
        exec(imports)
    except Exception as e:
        print("---import 체크: 실패---")
        error_message = [("user", f"당신의 코드는 import 테스트를 실패했습니다: {e}")]
        messages += error_message
        return {
            "generation": code_solution,
            "messages": messages,
            "iterations": iterations,
            "error": "yes",
        }

    try:
        exec(imports + "\n" + code)
    except Exception as e:
        print("---code block 체크: 실패---")
        error_message = [("user", f"당신의 코드는 실행 테스트를 실패했습니다: {e}")]
        messages += error_message
        return {
            "generation": code_solution,
            "messages": messages,
            "iterations": iterations,
            "error": "yes",
        }

    print("---오류 없음---")
    return {
        "generation": code_solution,
        "messages": messages,
        "iterations": iterations,
        "error": "no",
    }


def reflect(state: GraphState):
    """오류 반영"""
    print("---코드 솔루션 생성---")
    messages = state["messages"]
    iterations = state["iterations"]
    code_solution = state["generation"]
    reflections = code_gen_chain.invoke(
        {"context": concatenated_content, "messages": messages}
    )
    messages = [("assistant", f"여기 오류를 반영한 코드입니다: {reflections}")]
    return {"generation": code_solution, "messages": messages, "iterations": iterations}

code_check 는 검증을 import 단계 와 실행 단계 로 나눠서 검사한다 — 어느 단계에서 실패했는지가 구분돼야 LLM 에게 더 정확한 재시도 지시를 줄 수 있기 때문이다. 다음은 종료 여부를 결정하는 에지다. 오류가 없거나 최대 시도 횟수(3회)에 도달하면 종료하고, 그렇지 않으면 재시도한다.

flag = "do not reflect"

def decide_to_finish(state: GraphState):
    """종료 여부를 결정합니다."""
    error = state["error"]
    iterations = state["iterations"]
    if error == "no" or iterations == 3:
        print("---종료---")
        return "end"
    else:
        print("---재시도---")
        if flag is True:
            return "reflect"
        else:
            return "generate"

마지막으로 노드와 에지를 연결한다.

from langgraph.graph import END, StateGraph, START

workflow = StateGraph(GraphState)

workflow.add_node("generate", generate)
workflow.add_node("check_code", code_check)
workflow.add_node("reflect", reflect)

workflow.add_edge(START, "generate")
workflow.add_edge("generate", "check_code")
workflow.add_conditional_edges(
    "check_code",
    decide_to_finish,
    {
        "end": END,
        "reflect": "reflect",
        "generate": "generate",
    },
)
workflow.add_edge("reflect", "generate")

app = workflow.compile()

완성된 그래프를 테스트한다.

question = "문자열을 runnable 객체에 직접 전달하고, 이를 사용하여 내 프롬프트에 필요한 입력을 구성하려면 어떻게 해야 하나요?"
app.invoke({"messages": [("user", question)], "iterations": 0})

---코드 생성---
---코드 검사---
---오류 없음---
---종료---
{'error': 'no',
 'messages': [...],
 ...}

이렇게 제공받은 문서를 기반으로 코드를 생성하고 직접 실행해 검증하는 코드 어시스트 챗봇이 완성됐다.

핵심 개념 정리

실무 체크리스트

• [ ] 상태에 리스트 필드를 둘 때 Annotated[list, add] 같은 리듀서를 지정했는가 — 안 하면 새 값이 기존 리스트를 덮어쓴다
• [ ] 조건부 에지의 매핑 딕셔너리 키가 라우팅 함수의 실제 반환값과 정확히 일치하는가
• [ ] 체크포인터를 쓰는 그래프에서 대화(사용자)마다 서로 다른 thread_id 를 주고 있는가
• [ ] interrupt_before 로 멈춘 뒤 get_state().next 로 다음 실행 노드를 확인했는가
• [ ] 새 human-in-the-loop 기능을 만들 때 interrupt_before 대신 interrupt() + Command(resume=...) 를 검토했는가(최신 동향 참고)
• [ ] Corrective-RAG 의 grade_documents 가 문서 하나라도 무관하면 web_search 플래그를 세우는지 확인했는가
• [ ] 코드 어시스트 챗봇에서 iterations 상한을 두어 무한 재시도를 막았는가
• [ ] exec() 로 생성된 코드를 검증할 때 import 단계와 실행 단계의 오류를 각각 따로 잡고 있는가
• [ ] 프리빌트 컴포넌트(ToolNode·tools_condition)로 대체 가능한데 직접 구현 노드를 그대로 남겨 두지 않았는가

연습문제

1. 적용. 고객 지원 챗봇을 만드는데, 사용자가 "환불 요청"이라고 말하면 사람 상담원에게 넘기고, 그 외 질문은 LLM 이 바로 답하게 하고 싶다. 어떤 종류의 에지를 써야 하며, 그 에지를 만드는 데 필요한 것은 무엇인가?
2. 판단. 두 사용자가 동시에 같은 챗봇을 쓰는데, B 사용자의 질문에 A 사용자와 나눈 대화 내용이 섞여 나온다는 신고가 들어왔다. 어디를 먼저 점검해야 하는가?
3. 설계. RAG 챗봇이 검색한 문서가 질문과 무관할 때 그냥 "모른다"고 답하는 대신, 재작성한 질문으로 웹을 검색해 보완하고 싶다. 이 장의 어떤 그래프 구조를 참고해야 하며, 분기의 기준이 되는 상태 필드는 무엇인가?
4. 실무. LLM 이 생성한 코드를 사용자에게 보여주기 전에 자동으로 검증하고 싶다. 코드 어시스트 챗봇의 code_check 노드는 검증을 어떤 두 단계로 나누며, 왜 나누어 검사하는가?

최신 동향 (2026-09 기준)

최신 동향 (검증 2026-09-12) — 이 장의 실습 코드 세 곳이 책 출판 이후 상위 호환 경로로 바뀌었다. 그래프·상태·노드·에지 개념과 실습 흐름 자체는 그대로 유효하다. - Tavily 검색 도구: 본문이 쓰는 langchain_community.tools.tavily_search.TavilySearchResults 는 공식 레퍼런스에 Deprecated 로 표시돼 있다. 현재는 langchain-tavily 패키지의 TavilySearch 클래스가 권장 경로다(pip install -U langchain-tavily, from langchain_tavily import TavilySearch). - Pydantic 임포트 경로: GradeDocuments·code 모델이 쓰는 from langchain_core.pydantic_v1 import BaseModel, Field 는 pydantic v1 호환을 위한 임시 경로이며 langchain_core.utils.pydantic 참조 문서에 DEPRECATED 로 표시돼 있다. 신규 코드는 from pydantic import BaseModel, Field(pydantic v2)를 직접 쓰는 것이 권장된다. - 루프 개입 방식: §2.5 의 interrupt_before 는 여전히 동작하지만, LangGraph 공식 Interrupts 문서는 "정적 중단점(interrupt_before/interrupt_after)은 human-in-the-loop 워크플로우에 권장하지 않는다. interrupt() 함수를 대신 쓰라"고 명시한다. 현재 권장 패턴은 노드 안에서 interrupt(value) 를 호출해 실행을 멈추고, 사람의 입력을 Command(resume=값) 으로 그래프에 되돌려 재개하는 방식이다. interrupt_before 는 여전히 디버깅용 정적 중단점으로는 유효하다.

부록 A. 핵심 비교표

부록 B. 추천 참고 자료

외부 자료 (Tier 1 공식, 생존 확인 2026-09-12)

• LangGraph 공식 개요 문서 — LangGraph overview
• 프리빌트 컴포넌트 레퍼런스 — ToolNode·tools_condition 등
• Interrupts(루프 개입) 공식 가이드 — 정적 중단점과 interrupt() 비교
• Tavily 통합 공식 문서 — langchain-tavily 설치와 사용법
• 구글 프리겔 논문 — Pregel: A System for Large-Scale Graph Processing

본 책 연계 챕터

부록 C. 연습문제 풀이

1. (문제 1 정답) 조건부 에지(add_conditional_edges)를 써야 한다. 사용자의 발화를 판단해 "환불 요청"이면 사람 라우팅 노드로, 그 외에는 챗봇 노드로 보내는 라우팅 함수와, 그 함수의 반환값을 노드 이름에 매핑하는 딕셔너리가 필요하다(§1.4 조건부 에지).
2. (문제 2 정답) thread_id 설정을 먼저 점검한다. 두 사용자가 같은 thread_id 를 쓰면 체크포인터가 같은 스레드의 저장된 상태를 불러와 대화가 섞인다. 사용자별로 서로 다른 thread_id 를 config 에 넣고 있는지 확인한다(§2.4 상태 저장하기).
3. (문제 3 정답) 자체교정 RAG(Corrective-RAG) 구조를 참고한다(§3.1). 분기 기준이 되는 상태 필드는 web_search(웹 검색 필요 여부 플래그)이며, grade_documents 노드가 이 값을 세우고 decide_to_generate 조건부 에지가 이 값을 보고 transform_query(질문 재작성) 또는 generate(바로 답변)로 분기한다.
4. (문제 4 정답) import 단계와 코드 실행 단계로 나눠 검사한다. exec(imports) 로 먼저 import 문만 실행해 보고, 통과하면 exec(imports + "\n" + code) 로 전체 코드를 실행한다. 나누는 이유는 오류가 "import 가 잘못됐는지" "로직이 잘못됐는지" 를 구분해 LLM 에게 더 정확한 재시도 지시를 줄 수 있기 때문이다(§3.2 코드 어시스트 챗봇).